package com.atguigu.userprofile.task;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constant.ConstCode;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.util.ClickhouseUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 将Clickhouse中的标签宽表转换成bitmap表
 *
 * 任务步骤:
 * 0. 准备SparkSql环境（本任务跟Spark和Hive没关系， 只是为了任务的统一调度处理）
 *
 * 1. 获取外部传入的参数  taskId , busiDate
 *
 * 2. 明确宽表的列 ， 通过被计算的标签来确定
 *
 * 3. 创建最终的bitmap表
 *
 * 4. 组织SQL: insert(bitmap表) .... select(宽表)
 *
 * 5. 执行SQL
 *
 *
 */
public class TaskBitmap {

    public static void main(String[] args) {
        //0.准备SparkSql环境（本任务跟Spark和Hive没关系， 只是为了任务的统一调度处理）
        SparkConf sparkConf = new SparkConf().setAppName("task_bitmap_app");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //1. 获取外部传入的参数  taskId , busiDate
        String taskId = args[0] ;
        String busiDate = args[1] ;

        //2. 明确宽表的列 ， 通过被计算的标签来确定
        List<TagInfo> tagInfos = TagInfoDao.selectTagInfosWithTaskStatusEnable();

        //3. 创建最终的bitmap表
        // 最终的结果相对比较固定，不需要动态建表，提前将表建好
        // 因为标签值的类型有四种: 整型 浮点 文本 日期，
        // 所以最终我们需要创建4张bitmap表，分别存储不同类型的标签
        /*
        create table user_tag_value_string(
            tag_code String,
            tag_value String ,
            us AggregateFunction(groupBitmap,UInt64),
            dt String
        )engine=AggregatingMergeTree()
        partition by  (dt, tag_code)
        order by (tag_value);

        create table  user_tag_value_long(
            tag_code String,
            tag_value UInt64 ,
            us AggregateFunction(groupBitmap,UInt64),
            dt String
        )engine=AggregatingMergeTree()
        partition by  (dt, tag_code)
        order by (tag_value);

        create table  user_tag_value_decimal(
            tag_code String,
            tag_value Decimal64(2) ,
            us AggregateFunction(groupBitmap,UInt64),
            dt String
        )engine=AggregatingMergeTree()
        partition by  (dt, tag_code)
        order by (tag_value);

        create table  user_tag_value_date(
            tag_code String,
            tag_value Date ,
            us AggregateFunction(groupBitmap,UInt64),
            dt String
        )engine=AggregatingMergeTree()
        partition by  (dt, tag_code)
        order by (tag_value);

         */

        String longTableName = "user_tag_value_long" ;
        String decimalTableName ="user_tag_value_decimal" ;
        String stringTableName = "user_tag_value_string" ;
        String dateTableName = "user_tag_value_date" ;


        //4. 组织SQL: insert(bitmap表) .... select(宽表)

        //4.1 将计算的所有的标签按照标签值类型分成4份

        ArrayList<TagInfo> longTagInfos = new ArrayList<>();
        ArrayList<TagInfo> decimalTagInfos = new ArrayList<>();
        ArrayList<TagInfo> stringTagInfos = new ArrayList<>();
        ArrayList<TagInfo> dateTagInfos = new ArrayList<>();

        for (TagInfo tagInfo : tagInfos) {
            String tagValueType = tagInfo.getTagValueType();
            if(ConstCode.TAG_VALUE_TYPE_LONG.equals(tagValueType)){
                longTagInfos.add(tagInfo) ;
            }else if (ConstCode.TAG_VALUE_TYPE_DECIMAL.equals(tagValueType)){
                decimalTagInfos.add(tagInfo);
            }else if (ConstCode.TAG_VALUE_TYPE_STRING.equals(tagValueType)){
                stringTagInfos.add(tagInfo);
            }else if( ConstCode.TAG_VALUE_TYPE_DATE.equals(tagValueType)){
                dateTagInfos.add(tagInfo);
            }
        }


        //4.2  获取insert select SQL
        String longTypeSql = genInsertBitMapSql(longTagInfos, longTableName ,busiDate);
        String decimalTypeSql = genInsertBitMapSql(decimalTagInfos, decimalTableName, busiDate);
        String stringTypeSql = genInsertBitMapSql(stringTagInfos, stringTableName ,busiDate);
        String dateTypeSql = genInsertBitMapSql(dateTagInfos , dateTableName , busiDate) ;

        System.out.println("longTypeSql : " + longTypeSql);
        System.out.println("decimalTypeSql : " + decimalTypeSql);
        System.out.println("stringTypeSql : " + stringTypeSql);
        System.out.println("dateTypeSql : " + dateTypeSql);


        //5. 执行SQL
        //删除当前业务日期的数据
        deleteCurrentBusiDateData(longTableName , busiDate);
        deleteCurrentBusiDateData(decimalTableName , busiDate);
        deleteCurrentBusiDateData(stringTableName , busiDate);
        deleteCurrentBusiDateData(dateTableName , busiDate);

        //写入
        insertBitMapTable(longTypeSql);
        insertBitMapTable(decimalTypeSql);
        insertBitMapTable(stringTypeSql);
        insertBitMapTable(dateTypeSql);


    }

    /*
        insert into user_tag_value_bitmap
        select t1.tt.1 as tag_code , t1.tt.2  as tag_value  , groupBitmapState( t1.uid )  from
         (select uid ,
               arrayJoin ([ ('gender' ,  gender) , ('agegroup' , agegroup) , ('favor' , favor) ])  as tt
        from user_tag_merge ) t1
        group by t1.tt.1 , t1.tt.2


         ('favor' , favor)   => ('favor' , if(favor == '' , '0' , favor))

     */

    private static String  genInsertBitMapSql( List<TagInfo> tagInfos , String bitmapTableName , String busiDate ){

        //先判断当前集合中是否有标签
        if(tagInfos.isEmpty()){
            return null ;
        }
        //动态的将arrayJoin中的内容处理好

        //TODO 判断标签值是否为空
        // 如果为空，
        // long和decimal类型的统一赋默认值为0
        // date类型的统一赋默认值为 '1970-01-01'     formatDateTime(cast('1970-01-01' as Date) , '%F')
        // 如上的默认值处理一定要考虑实际情况，不是标准答案.
        String arrayJoinSql = null ;
        if(bitmapTableName.equals("user_tag_value_long") || bitmapTableName.equals("user_tag_value_decimal")){
            //long 或者 decimal => 0
            arrayJoinSql =  tagInfos.stream().map(
                    tagInfo -> "( '" + tagInfo.getTagCode().toLowerCase() + "' , if(" + tagInfo.getTagCode().toLowerCase() + "=='' , '0' , "+ tagInfo.getTagCode().toLowerCase() +") )"
            ).collect(Collectors.joining(" , "));
        }else if (bitmapTableName.equals("user_tag_value_date")){
            //date => '1970-01-01'
            arrayJoinSql =  tagInfos.stream().map(
                    tagInfo -> "( '" + tagInfo.getTagCode().toLowerCase() + "' , if(" + tagInfo.getTagCode().toLowerCase() + "=='', '1970-01-01', " + tagInfo.getTagCode().toLowerCase() +") )"
            ).collect(Collectors.joining(" , "));
        }else{
            //string 不做处理
            arrayJoinSql =  tagInfos.stream().map(
                    tagInfo -> "( '" + tagInfo.getTagCode().toLowerCase() + "' , " + tagInfo.getTagCode().toLowerCase() + " )"
            ).collect(Collectors.joining(" , "));
        }

        String selectTableName = "up_merge_" + busiDate.replace("-" , "") ;

        String insertSelectSql = " insert into "  +  bitmapTableName +
                " select t1.tt.1 as tag_code , t1.tt.2  as tag_value  , groupBitmapState(cast( t1.uid as UInt64))  , '"+ busiDate +"' from " +
                " ( select uid , arrayJoin ([ " + arrayJoinSql + " ]) as tt from "+ selectTableName +" ) t1 " +
                " group by t1.tt.1 , t1.tt.2 " ;


        return insertSelectSql ;
    }

    private static void deleteCurrentBusiDateData(String bitmapTableName , String busiDate){
        String deleteSql = "alter table " + bitmapTableName + " delete where dt = '" + busiDate + "'" ;
        ClickhouseUtil.executeSql(deleteSql);
    }


    private static void  insertBitMapTable(String insertSelectSql){
        if(insertSelectSql != null ){
            ClickhouseUtil.executeSql(insertSelectSql);
        }
    }
}
